package com.tencent.sr.iris.activity.service.activity.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.tencent.sr.iris.activity.common.enums.TaskRecordAwardIssueStatusEnum;
import com.tencent.sr.iris.activity.domain.event.dto.EsQueryDTO;
import com.tencent.sr.iris.activity.interfaces.dto.MultiTaskActivityDTO;
import com.tencent.sr.iris.activity.interfaces.dto.TargetThresholdVO;
import com.tencent.sr.iris.activity.interfaces.enums.ActivityIdEnum;
import com.tencent.sr.iris.activity.interfaces.enums.InviteGiftVersionEnum;
import com.tencent.sr.iris.activity.interfaces.enums.IsParticipatedEnum;
import com.tencent.sr.iris.activity.interfaces.request.IrisPaginationRequest;
import com.tencent.sr.iris.activity.interfaces.request.acticity.BatchDataRequest;
import com.tencent.sr.iris.activity.interfaces.response.SearchAfterPaginationResponse;
import com.tencent.sr.iris.activity.interfaces.response.activity.BatchDataResponse;
import com.tencent.sr.iris.activity.repository.entity.TIrisActivityTaskDO;
import com.tencent.sr.iris.activity.repository.entity.TIrisUserTaskBatchDataDO;
import com.tencent.sr.iris.activity.repository.singlemapper.example.TIrisUserTaskBatchDataExample;
import com.tencent.sr.iris.activity.service.activity.ActivityCommonService;
import com.tencent.sr.iris.activity.service.activity.BatchDataQueryService;
import com.tencent.sr.iris.activity.service.activity.EsDataOperateService;
import com.tencent.sr.iris.activity.service.activity.IrisActivityTaskService;
import com.tencent.sr.iris.activity.service.config.ActivityConfig;
import com.tencent.sr.iris.activity.service.convert.UserTaskBatchDataConvert;
import com.tencent.sr.iris.activity.service.event.util.ActivityCommonUtilService;
import com.tencent.sr.iris.activity.service.util.ESUtils;
import com.tencent.sr.iris.activity.service.util.UserTaskBatchDataUtils;
import com.tencent.sr.rmall.common.response.PaginationResponse;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedScriptedMetric;
import org.elasticsearch.search.collapse.CollapseBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Service
public class IrisEsBatchDataQueryServiceImpl extends EsDataOperateService implements BatchDataQueryService {

    @Resource
    private UserTaskBatchDataConvert batchDataConvert;

    @Resource
    private ActivityConfig activityConfig;

    @Resource
    private IrisActivityTaskService irisActivityTaskService;

    @Resource
    private ActivityCommonService activityCommonService;

    @Resource
    private ActivityCommonUtilService activityCommonUtilService;

    @Value("${env.es.batch.data.index.name:user-task-batch-data}")
    private String indexName;

    @Override
    protected String[] indices() {
        return new String[]{indexName};
    }

    @Override
    public SearchAfterPaginationResponse<BatchDataResponse> page(IrisPaginationRequest<BatchDataRequest> request) {
        MultiTaskActivityDTO multiTaskActivityDTO = activityConfig.getMultiTaskActivityConfig()
                .get(Optional.ofNullable(request.getQueryCriteria().getActivityId())
                        .map(ActivityIdEnum::getCode).orElse(request.getQueryCriteria().getActivityType().getCode()));
        if (multiTaskActivityDTO != null) {
            return searchMultiTask(request);
        }

        return searchSingle(request);
    }

    /**
     * 单任务搜索
     *
     * @param request
     * @return
     */
    private SearchAfterPaginationResponse<BatchDataResponse> searchSingle(
            IrisPaginationRequest<BatchDataRequest> request) {

        BatchDataRequest queryCriteria = request.getQueryCriteria();
        String searchAfter = queryCriteria.getSearchAfter();
        boolean bSearchAfter = StrUtil.isNotBlank(searchAfter);

        Integer pageSize = request.getPageSize();
        Integer pageNum = bSearchAfter ? 1 : request.getPageNum();

        BoolQueryBuilder query = QueryBuilders.boolQuery();
        UserTaskBatchDataUtils.buildCondition(query, queryCriteria, null);

        AggregationBuilder aggregation = null;
        if (bSearchAfter) {
            query.must(QueryBuilders.rangeQuery(TIrisUserTaskBatchDataExample.UID).gt(searchAfter));
        } else {
            aggregation = getPreciseCountAggregation(TIrisUserTaskBatchDataExample.UID, null);
        }

        FieldSortBuilder sortBuilder = SortBuilders.fieldSort(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.UID))
                .order(SortOrder.ASC);
        EsQueryDTO esQueryDTO = EsQueryDTO.builder().query(query).routing("0").totalAggregation(aggregation)
                .pageNum(pageNum).pageSize(pageSize).sortBuilder(sortBuilder).build();

        Function<SearchResponse, List<TIrisUserTaskBatchDataDO>> func = (r) -> ESUtils
                .buildResponse(r, TIrisUserTaskBatchDataDO.class);
        PaginationResponse<TIrisUserTaskBatchDataDO> searchPage = searchPage(esQueryDTO, func);

        SearchAfterPaginationResponse<BatchDataResponse> response = new SearchAfterPaginationResponse<>();
        response.setDataList(Collections.emptyList());
        response.setTotalCount(0);
        response.setPageNum(pageNum);
        response.setPageSize(pageSize);
        List<TIrisUserTaskBatchDataDO> dataList = searchPage.getDataList();
        if (CollectionUtil.isEmpty(dataList)) {
            return response;
        }

        List<BatchDataResponse> responses = dataList.stream()
                .map(batchDataConvert::buildFromBatchDataDO).collect(Collectors.toList());
        handleThreshold(responses);
        response.setTotalCount(searchPage.getTotalCount());
        response.setDataList(responses);

        handleLimit(responses, queryCriteria);
        return response;
    }

    /**
     * 处理门槛值
     *
     * @param responses
     */
    private void handleThreshold(List<BatchDataResponse> responses) {
        if (CollectionUtil.isNotEmpty(responses)) {

            List<String> taskIds = responses.stream()
                    .map(BatchDataResponse::getTaskId).collect(Collectors.toList());
            List<TIrisActivityTaskDO> tasks = irisActivityTaskService.getTasksByIds(taskIds);
            Map<String, TIrisActivityTaskDO> taskMap = tasks.stream()
                    .collect(Collectors.toMap(i -> String.valueOf(i.getBizId()), i -> i, (i1, i2) -> i1));

            Map<String, MultiTaskActivityDTO> multiTaskActivityConfig = activityConfig.getMultiTaskActivityConfig();

            for (BatchDataResponse res : responses) {
                String uid = res.getUid();

                MultiTaskActivityDTO multiTaskActivityDTO = multiTaskActivityConfig.get(res.getActivityType());
                String groupByExpression = Optional.ofNullable(multiTaskActivityDTO)
                        .map(MultiTaskActivityDTO::getGroupByExpression).orElse(null);

                String taskId = res.getTaskId();
                TIrisActivityTaskDO taskDO = Optional.ofNullable(taskMap.get(taskId))
                        .orElseGet(TIrisActivityTaskDO::new);
                //门槛
                String conditionThreshold = taskDO.getConditionThreshold();

                TargetThresholdVO threshold = Optional.ofNullable(JSON.parseObject(
                        conditionThreshold, TargetThresholdVO.class)).orElseGet(TargetThresholdVO::new);

                res.setTotalCommission(threshold.getTargetCommission())
                        .setTotalShareCount(threshold.getTargetShare())
                        .setGroupByExpression(groupByExpression)
                        .setCommissionTaskPlusDay(activityConfig.getCommissionPlusDays(uid));
            }

        }
    }

    /**
     * 多阶段任务搜索
     *
     * @param request
     * @return
     */
    private SearchAfterPaginationResponse<BatchDataResponse> searchMultiTask(
            IrisPaginationRequest<BatchDataRequest> request) {

        BatchDataRequest queryCriteria = request.getQueryCriteria();
        String searchAfter = queryCriteria.getSearchAfter();
        boolean bSearchAfter = StrUtil.isNotBlank(searchAfter);

        Integer pageSize = request.getPageSize();
        Integer pageNum = bSearchAfter ? 1 : request.getPageNum();

        AggregationBuilder aggregation = null;
        BoolQueryBuilder query = QueryBuilders.boolQuery();
        if (bSearchAfter) {
            query.must(QueryBuilders.rangeQuery(TIrisUserTaskBatchDataExample.UID).gt(searchAfter));
        } else {
            aggregation = getPreciseCountAggregation(TIrisUserTaskBatchDataExample.UID, null);
        }
        UserTaskBatchDataUtils.buildCondition(query, request.getQueryCriteria(), null);

        //先根据uid去重分页查询
        FieldSortBuilder sortBuilder = SortBuilders.fieldSort(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.UID))
                .order(SortOrder.ASC);
        EsQueryDTO esQueryDTO = EsQueryDTO.builder().query(query).routing("0")
                .pageNum(pageNum).pageSize(pageSize).sortBuilder(sortBuilder).totalAggregation(aggregation)
                .collapseBuilder(new CollapseBuilder(TIrisUserTaskBatchDataExample.UID)).build();
        Function<SearchResponse, List<String>> func = (r) -> ESUtils
                .getFieldList(r, TIrisUserTaskBatchDataExample.UID, String.class);

        SearchAfterPaginationResponse<String> searchPage = searchPage(esQueryDTO, func);

        SearchAfterPaginationResponse<BatchDataResponse> response = new SearchAfterPaginationResponse<>();
        response.setTotalCount(0);
        response.setDataList(Collections.emptyList());
        response.setPageNum(pageNum);
        response.setPageSize(pageSize);

        List<String> uidList = searchPage.getDataList();
        if (CollectionUtil.isEmpty(uidList)) {
            return response;
        }
        response.setTotalCount(Math.toIntExact(searchPage.getTotalCount()));

        //查询uid下的相同条件的所有任务
        query = QueryBuilders.boolQuery();
        UserTaskBatchDataUtils.buildCondition(query, request.getQueryCriteria(), uidList);
        esQueryDTO = EsQueryDTO.builder().query(query).routing("0")
                .pageNum(1).pageSize(esConfig.getMaxPageSize()).sortBuilder(sortBuilder).build();
        Function<SearchResponse, List<TIrisUserTaskBatchDataDO>> secondFunc = (r) -> ESUtils
                .buildResponse(r, TIrisUserTaskBatchDataDO.class);
        PaginationResponse<TIrisUserTaskBatchDataDO> page = searchPage(esQueryDTO, secondFunc);

        List<BatchDataResponse> responses = page.getDataList().stream()
                .map(batchDataConvert::buildFromBatchDataDO).collect(Collectors.toList());

        handleThreshold(responses);
        response.setDataList(responses);

        handleLimit(responses, queryCriteria);
        return response;
    }

    @Override
    public long countDistinctUid(String pid, long monthTimestamp, String activityType, String activityId, List<String> taskId) {
        DateTime date = DateUtil.date(monthTimestamp);
        DateTime begin = DateUtil.beginOfMonth(date);
        DateTime end = DateUtil.endOfMonth(date);
        BoolQueryBuilder query = QueryBuilders.boolQuery()
                .must(QueryBuilders.termQuery(TIrisUserTaskBatchDataExample.PID, pid))
                .must(QueryBuilders.termQuery(TIrisUserTaskBatchDataExample.ACTIVITY_TYPE, activityType))
                .must(QueryBuilders.termQuery(TIrisUserTaskBatchDataExample.ACTIVITY_ID, activityId))
                .must(QueryBuilders.termsQuery(TIrisUserTaskBatchDataExample.TASK_ID, taskId))
                .must(QueryBuilders.rangeQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.CREATE_TIME))
                        .gte(begin.getTime()).lte(end.getTime()));
        return esCountDistinctPrecise(query, TIrisUserTaskBatchDataExample.UID, "0", esConfig.getMaxPreciseCount());
    }

    private void handleLimit(List<BatchDataResponse> responseList, BatchDataRequest request) {
        if (CollectionUtil.isEmpty(responseList)) {
            return;
        }

        List<BatchDataResponse> limitedResponseList = responseList.stream()
                .filter(i -> CollectionUtil.contains(activityConfig.getLimitedActivity().get(i.getActivityId()), i.getTaskId()))
                .collect(Collectors.toList());
        List<String> pidList = limitedResponseList.stream().map(BatchDataResponse::getPid)
                .filter(StrUtil::isNotBlank).collect(Collectors.toList());
        List<Date> dateList = limitedResponseList.stream().filter(i -> i.getCurrentTaskStartTime() != null)
                .map(i -> DateUtil.date(i.getCurrentTaskStartTime()))
                .collect(Collectors.toList());
        //无受限pid
        if (CollectionUtil.isEmpty(pidList)) {
            return;
        }

        List<String> limitedTaskIds = limitedResponseList.stream()
                .map(BatchDataResponse::getTaskId).distinct().collect(Collectors.toList());
        //查询已贡献
        Map<String, Integer> contributedCountMap = getContributedCountMap(pidList, limitedTaskIds, request);

        Map<String, Map<String, Integer>> inviteLimitMap = activityCommonService.queryInviteLimitMap(pidList, dateList);

        responseList.stream()
                .filter(i -> CollectionUtil.contains(activityConfig.getLimitedActivity().get(i.getActivityId()), i.getTaskId()))
                .forEach(i -> {
                    //上级实时限额数
                    String monthStr = DateUtil.format(DateUtil.date(i.getCurrentTaskStartTime()), "yyyyMM");
                    Integer inviteLimit = inviteLimitMap.getOrDefault(monthStr, Collections.emptyMap()).get(i.getPid());
                    i.setMonthInviteLimit(inviteLimit);
                    //已占用额度
                    Integer used = contributedCountMap.getOrDefault(i.getPid(), 0);
                    i.setUsedInviteLimit(used);
                });
    }

    /**
     * 获取已贡献数
     * @param request
     * @param pidList
     * @return
     */
    private Map<String, Integer> getContributedCountMap(List<String> pidList, List<String> limitTaskIds, BatchDataRequest request) {
        BoolQueryBuilder query = QueryBuilders.boolQuery()
                .must(QueryBuilders.termsQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.PID), pidList))
                .must(QueryBuilders.termQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.IS_PARTICIPATED), IsParticipatedEnum.YES.name()))
                .must(QueryBuilders.termQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.ACTIVITY_TYPE), request.getActivityType().getCode()))
                .must(QueryBuilders.termsQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.TASK_ID), limitTaskIds))
                .must(QueryBuilders.termQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.ACTIVITY_ID), request.getActivityId().getCode()));
        Long startTime = request.getStartTime();
        Long endTime = request.getEndTime();
        if (startTime != null && endTime != null) {
            Date end = DateUtil.endOfMonth(DateUtil.date(endTime));
            Date start = DateUtil.beginOfMonth(DateUtil.date(startTime));
            query.must(QueryBuilders.rangeQuery(
                            StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.CURRENT_TASK_START_TIME)).gte(start.getTime())
                    .lte(end.getTime()));
            InviteGiftVersionEnum inviteGiftVersion = activityCommonUtilService.getInviteGiftVersion(false, startTime, endTime);
            if (Objects.equals(inviteGiftVersion, InviteGiftVersionEnum.V0706)) {
                query.must(QueryBuilders.termsQuery(StrUtil.toCamelCase(TIrisUserTaskBatchDataExample.AWARD_ISSUE_STATUS),
                        Arrays.asList(TaskRecordAwardIssueStatusEnum.ISSUE_SUCCEED.getCode())));
            }
        }

        TermsAggregationBuilder termsAgg = AggregationBuilders.terms("pidGroup").minDocCount(1)
                .field(TIrisUserTaskBatchDataExample.PID).size(esConfig.getTermsSize()).shardSize(esConfig.getTermsSize());
        AggregationBuilder countAggregation = getPreciseCountAggregation(TIrisUserTaskBatchDataExample.UID, esConfig.getMaxPreciseCount());
        termsAgg.subAggregation(countAggregation);
        EsQueryDTO esQueryDTO = EsQueryDTO.builder().query(query).routing("0").totalAggregation(termsAgg)
                .pageNum(0).pageSize(0).build();
        SearchResponse searchResponse = getSearchResponse(esQueryDTO);

        Map<String, Integer> contributedCountMap = Maps.newHashMap();
        Aggregations aggregations = searchResponse.getAggregations();
        ParsedStringTerms parsedStringTerms = aggregations.get(termsAgg.getName());
        if (parsedStringTerms != null && CollectionUtil.isNotEmpty(parsedStringTerms.getBuckets())) {
            List<Terms.Bucket> buckets = (List<Terms.Bucket>) parsedStringTerms.getBuckets();
            for (Terms.Bucket bucket : buckets) {
                String pid = bucket.getKeyAsString();
                ParsedScriptedMetric metric = bucket.getAggregations().get(countAggregation.getName());
                contributedCountMap.put(pid, (Integer) metric.aggregation());
            }
        }
        return contributedCountMap;
    }
}
